package com.etl.extract;

import com.etl.enums.ScheduleStatusEnum;
import com.etl.exception.TaskBeanException;
import com.etl.sys.entity.SysSchedule;
import com.etl.sys.service.ISysLockService;
import com.etl.sys.service.ISysScheduleService;
import com.etl.task.basic.ITask;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.scheduling.support.CronTrigger;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledFuture;
import java.util.stream.Collectors;

/**
 * @author: chenfenghai
 * @version: 1.0
 * @date: 2020-10-21 10:22
 */
@Slf4j
@Component
public class ScheduleBuilder {

    @Autowired
    TaskBuilder taskBuilder;
    @Autowired
    ScheduleTaskBuilder scheduleTaskBuilder;
    @Autowired
    protected DatabaseBuilder databaseBuilder;
    @Autowired
    private ThreadPoolTaskScheduler threadPoolTaskScheduler;
    @Autowired
    ISysScheduleService sysScheduleService;

    /**
     * 调度ID和调度结果KEY-VALUE键值对，保存调度结果
     */
    private static final Map<Long, ScheduledFuture<?>>
            SCHEDULED_FUTURE_MAP = new ConcurrentHashMap<>();

    /**
     * 加载调度定时器
     */
    public void loadScheduleTimer() {
        log.debug("Load schedule timer starting...");
        List<SysSchedule> scheduleList = sysScheduleService.lambdaQuery()
                .eq(SysSchedule::getLockUid, ISysLockService.LOCK_UID).list();
        cancelScheduleDeleted(scheduleList.stream().map(SysSchedule::getScheduleId)
                .collect(Collectors.toList()));
        scheduleList.parallelStream().forEach(this::loadSchedule);
        log.debug("Load schedule timer completed.");
    }

    /**
     * 取消已删除的调度任务
     *
     * @param scheduleIdList 未删除的调度任务ID列表
     */
    private void cancelScheduleDeleted(List<Long> scheduleIdList) {
        SCHEDULED_FUTURE_MAP.entrySet().parallelStream()
                .filter(entry -> !scheduleIdList.contains(entry.getKey()))
                .forEach(entry -> {
                    log.info("Cancel schedule id: {}", entry.getKey());
                    SCHEDULED_FUTURE_MAP.remove(entry.getKey()).cancel(true);
                });
    }

    /**
     * 加载调度任务
     *
     * @param sysSchedule 调度记录
     */
    public void loadSchedule(SysSchedule sysSchedule) {
        try {
            scheduleTaskBuilder.buildTaskSchedule(sysSchedule);
        } catch (Exception ex) {
            log.error("Build TaskSchedule error: {}", sysSchedule);
            log.error("Build TaskSchedule error: ", ex);
            // 更新调度为已停止、已失败
            sysScheduleService.updateScheduleToStoppedAndFailed(sysSchedule.getScheduleId());
        }
    }

    /**
     * 刷新调度任务
     *
     * @param scheduleTask 调度任务
     */
    public void refreshSchedule(ScheduleTask scheduleTask) {
        ScheduledFuture<?> future = SCHEDULED_FUTURE_MAP.get(
                scheduleTask.getSchedule().getScheduleId());
        if (Objects.nonNull(future) && future.cancel(true)) {
            SCHEDULED_FUTURE_MAP.remove(scheduleTask.getSchedule().getScheduleId());
        }
        future = createScheduledFuture(scheduleTask);
        SCHEDULED_FUTURE_MAP.put(scheduleTask.getSchedule().getScheduleId(), future);
        // 更新调度状态为等待中
        sysScheduleService.updateScheduleStatus(
                scheduleTask.getSchedule().getScheduleId(), ScheduleStatusEnum.WAITING);
    }

    /**
     * 创建调度结果
     *
     * @param scheduleTask 调度任务
     * @return ScheduledFuture
     */
    private ScheduledFuture<?> createScheduledFuture(ScheduleTask scheduleTask) {
        Optional<ITask> optionalTask = taskBuilder.getTask(scheduleTask.getTask().getTaskCode());
        return threadPoolTaskScheduler.schedule(
                () -> optionalTask.orElseThrow(TaskBeanException::new).execute(scheduleTask),
                new CronTrigger(scheduleTask.getSchedule().getTimeCron()));
    }
}
